using System;
using System.Collections;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Hector.Api.Exceptions;

namespace Me.Prettyprint.Cassandra.Service
{
    /// <summary>
    /// When nectar gets the MBean data from Cassandra server, it is encapsulated
    /// by this class. Normally, developers wouldn't be calling Cassandra to get
    /// this information. It's more for performance monitoring or debugging.
    /// </summary>
    public class CassandraClientMonitor : CassandraClientMonitorMBean
    {
        private static readonly Logger log = new Logger(typeof(CassandraClientMonitor));
        public enum Counter
        {
            RECOVERABLE_TIMED_OUT_EXCEPTIONS,
            RECOVERABLE_UNAVAILABLE_EXCEPTIONS,
            RECOVERABLE_TRANSPORT_EXCEPTIONS,
            SKIP_HOST_SUCCESS,
            WRITE_SUCCESS,
            WRITE_FAIL,
            READ_SUCCESS,
            READ_FAIL,
            POOL_EXHAUSTED,
            RECOVERABLE_LB_CONNECT_ERRORS,
            CONNECT_ERROR,
            RENEWED_IDLE_CONNECTIONS,
            RENEWED_TOO_LONG_CONNECTIONS
        };
        private Dictionary<Counter, AtomicLong> counters = new Dictionary<Counter, AtomicLong>();
        private HConnectionManager connectionManager;

        public CassandraClientMonitor(HConnectionManager connectionManager)
        {
            this.connectionManager = connectionManager;
            foreach (Counter c in Enum.GetValues(typeof(Counter)))
            {
                counters.Add(c, new AtomicLong(0L));
            }
        }
        public long WriteSuccess()
        {
            return counters[Counter.WRITE_SUCCESS].Value;
        }
        public long ReadFail()
        {
            return counters[Counter.READ_FAIL].Value;
        }
        public long ReadSuccess()
        {
            return counters[Counter.READ_SUCCESS].Value;
        }
        public long SkipHostSuccess()
        {
            return counters[Counter.SKIP_HOST_SUCCESS].Value;
        }
        public long RecoverableTimedOutCount()
        {
            return counters[Counter.RECOVERABLE_TIMED_OUT_EXCEPTIONS].Value;
        }
        public long RecoverableUnavailableCount()
        {
            return counters[Counter.RECOVERABLE_UNAVAILABLE_EXCEPTIONS].Value;
        }
        public long RecoverableTransportExceptionCount()
        {
            return counters[Counter.RECOVERABLE_TRANSPORT_EXCEPTIONS].Value;
        }
        public long RecoverableLoadBalancedConnectErrors()
        {
            return counters[Counter.RECOVERABLE_LB_CONNECT_ERRORS].Value;
        }
        public long WriteFail()
        {
            return counters[Counter.WRITE_FAIL].Value;
        }
        public long NumConnectionErrors()
        {
            return counters[Counter.CONNECT_ERROR].Value;
        }
        public long NumPoolExhaustedEventCount()
        {
            return counters[Counter.POOL_EXHAUSTED].Value;
        }
        public int NumRenewedIdleConnections()
        {
            return (int)counters[Counter.RENEWED_IDLE_CONNECTIONS].Value;
        }
        public int NumRenewedTooLongConnections()
        {
            return (int)counters[Counter.RENEWED_TOO_LONG_CONNECTIONS].Value;
        }
        public void UpdateKnownHosts()
        {
            connectionManager.DoAddNodes();
        }
        public List<string> ExhaustedPoolNames()
        {
            List<string> downHosts = new List<string>();
            foreach (CassandraHost host in connectionManager.GetDownedHost())
            {
            }
            return downHosts;
        }
        public int NumActive()
        {
            int ret = 0;
            List<HClientPool> pools = connectionManager.ActivePools();
            foreach (HClientPool concurrentHClientPool in pools)
            {
                ret += concurrentHClientPool.NumActive;
            }
            return ret;
        }
        public int NumBlockedThreads()
        {
            int ret = 0;
            List<HClientPool> pools = connectionManager.ActivePools();
            foreach (HClientPool concurrentHClientPool in pools)
            {
                ret += concurrentHClientPool.NumBlockedThreads;
            }
            return ret;
        }
        public int NumExhaustedPools()
        {
            return connectionManager.GetDownedHost().Count;
        }
        public int NumIdleConnections()
        {
            int ret = 0;
            List<HClientPool> pools = connectionManager.ActivePools();
            foreach (HClientPool concurrentHClientPool in pools)
            {
                ret += concurrentHClientPool.NumIdle;
            }
            return ret;
        }
        public int NumPools()
        {
            return connectionManager.GetHosts().Count;
        }
        public List<string> KnownHosts()
        {
            List<string> hosts = new List<string>();
            foreach (CassandraHost cassandraHost in connectionManager.GetHosts())
            {
                hosts.Add(cassandraHost.ToString());
            }
            return hosts;
        }
        public List<string> StatisticsPerPool()
        {
            return connectionManager.StatusPerPool();
        }
        public long RecoverableErrorCount()
        {
            return RecoverableTimedOutCount() + RecoverableTransportExceptionCount() +
                RecoverableUnavailableCount() + RecoverableLoadBalancedConnectErrors();
        }
        public bool AddCassandraHost(String hostStr)
        {
            return connectionManager.AddCassandraHost(new CassandraHost(hostStr));
        }
        public bool RemoveCassandraHost(String hostStr)
        {
            return connectionManager.RemoveCassandraHost(new CassandraHost(hostStr));
        }
        public List<string> GetSuspendedCassandraHosts()
        {
            List<CassandraHost> hosts = connectionManager.SuspendedCassandraHosts();
            List<string> hostsStr = new List<string>();
            foreach (CassandraHost host in hosts)
            {
                hostsStr.Add(host.Name);    
            }
            return hostsStr;
        }
        public bool SuspendCassandraHost(String hostStr)
        {
            return connectionManager.SuspendCassandraHost(new CassandraHost(hostStr));
        }

        public bool UnsuspendCassandraHost(String hostStr)
        {
            return connectionManager.UnsuspendCassandraHost(new CassandraHost(hostStr));
        }
        public bool SetCassandraHostRetryDelay(String retryDelay)
        {
            int delay;
            try
            {
                delay = int.Parse(retryDelay);
                if (delay > 0)
                {
                    connectionManager.CassandraHostRetryDelay = delay;
                    return true;
                }
                else
                {
                    return false;
                }
            }
            catch (Exception e)
            {
                log.Error(e);
                return false;
            }
        }
        public void IncrementCounter(Counter c)
        {
            this.counters[c].Increment();
        }
        public void DecrementCounter(Counter c)
        {
            this.counters[c].Decrement();
        }
    }
}